/*
 * Licensed under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.submarine.registry;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.submarine.common.Constants;
import org.submarine.common.Logger;
import org.submarine.common.NamedThreadFactory;
import org.submarine.common.URL;
import org.submarine.registry.listener.NotifyListener;

/**
 * @file org.submarine.registry.AbstractRegistry.java
 *
 * AbstractRegistry
 * 
 * @author gaojian
 * @created 2013-8-18 上午12:15:24
 * @version 0.0.1
 * @since 0.0.1
 * 
 */
public abstract class AbstractRegistry {
	
	private Logger logger = Logger.getLogger(AbstractRegistry.class);
	
    private final ScheduledExecutorService retryExecutor = Executors.newScheduledThreadPool(1, new NamedThreadFactory("RegistryFailedRetryThread", true));

    private final ScheduledFuture<?> retryFuture;
    
    private volatile URL failedRegisteredURL;

    private volatile URL failedUnregisteredURL;
    
    private final ConcurrentMap<URL, Set<NotifyListener>> failedSubscribed = new ConcurrentHashMap<URL, Set<NotifyListener>>();

    private final ConcurrentMap<URL, Set<NotifyListener>> failedUnsubscribed = new ConcurrentHashMap<URL, Set<NotifyListener>>();
    
    private final ConcurrentMap<URL, Map<NotifyListener, Set<URL>>> failedNotified = new ConcurrentHashMap<URL, Map<NotifyListener, Set<URL>>>();

    private final ConcurrentMap<URL, Map<String, Set<URL>>> notified = new ConcurrentHashMap<URL, Map<String, Set<URL>>>();
    
    /**
     * 构造器
     */
    public AbstractRegistry() {
        this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() {
            public void run() {
                // 检测并连接注册中心
                try {
                    retry();
                } catch (Throwable t) { // 防御性容错
                    logger.record("Unexpected error occur at failed retry, cause: " + t.getMessage(), t);
                }
            }
        }, Constants.DEFAULT_REGISTRY_RETRY_PERIOD, Constants.DEFAULT_REGISTRY_RETRY_PERIOD, TimeUnit.MILLISECONDS);
    }
    
    /**
     * 
     * 由具体实现类执行注册
     *
     * @author gaojian
     * @created 2013-8-18 上午1:49:38
     *
     * @param url
     */
    protected abstract void doRegister(URL url);
    
    /**
     * 
     * 由具体实现类解除注册
     *
     * @author gaojian
     * @created 2013-8-18 上午1:49:59
     *
     * @param url
     */
    protected abstract void doUnregister(URL url);
    
    /**
     * 
     * 由具体实现类执行订阅
     *
     * @author gaojian
     * @created 2013-8-18 上午1:50:12
     *
     * @param url
     * @param listener
     */
    protected abstract void doSubscribe(URL url, NotifyListener listener);

    /**
     * 
     * 由具体实现类解除订阅
     *
     * @author gaojian
     * @created 2013-8-18 上午1:50:22
     *
     * @param url
     * @param listener
     */
    protected abstract void doUnsubscribe(URL url, NotifyListener listener);
    
    public abstract Set<URL> getInitCollectorURL();
    
    /**
     * 
     * 注册URL
     *
     * @author gaojian
     * @created 2013-8-18 上午1:45:59
     *
     * @param url
     */
    public void register(URL url) {
    	failedRegisteredURL = null;
        failedUnregisteredURL = null;
        try {
            doRegister(url);
        } catch (Exception e) {
            logger.record("Failed to register " + url + ", waiting for retry, cause: " + e.getMessage(), e);
            // 将失败的注册请求记录到失败列表，定时重试
            failedRegisteredURL = url;
        }
    }

    /**
     * 
     * 解除注册URL
     *
     * @author gaojian
     * @created 2013-8-18 上午1:46:12
     *
     * @param url
     */
    public void unregister(URL url) {
    	failedRegisteredURL = null;
        failedUnregisteredURL = null;
        try {
            // 向服务器端发送取消注册请求
            doUnregister(url);
        } catch (Exception e) {
            logger.record("Failed to uregister " + url + ", waiting for retry, cause: " + e.getMessage(), e);
            // 将失败的取消注册请求记录到失败列表，定时重试
            failedUnregisteredURL = url;
        }
    }
    
    /**
     * 
     * 移除订阅失败
     *
     * @author gaojian
     * @created 2013-8-18 上午1:46:31
     *
     * @param url
     * @param listener
     */
    private void removeFailedSubscribed(URL url, NotifyListener listener) {
        Set<NotifyListener> listeners = failedSubscribed.get(url);
        if (listeners != null) {
            listeners.remove(listener);
        }
        listeners = failedUnsubscribed.get(url);
        if (listeners != null) {
            listeners.remove(listener);
        }
        Map<NotifyListener, Set<URL>> notified = failedNotified.get(url);
        if (notified != null) {
            notified.remove(listener);
        }
    }
    
    /**
     * 
     * 添加订阅失败
     *
     * @author gaojian
     * @created 2013-8-18 上午1:46:52
     *
     * @param url
     * @param listener
     */
    private void addFailedSubscribed(URL url, NotifyListener listener) {
        Set<NotifyListener> listeners = failedSubscribed.get(url);
        if (listeners == null) {
            failedSubscribed.putIfAbsent(url, new HashSet<NotifyListener>());
            listeners = failedSubscribed.get(url);
        }
        listeners.add(listener);
    }
    
    /**
     * 
     * 订阅
     *
     * @author gaojian
     * @created 2013-8-18 上午1:47:07
     *
     * @param url
     * @param listener
     */
    public void subscribe(URL url, NotifyListener listener) {
        removeFailedSubscribed(url, listener);
        try {
            // 向服务器端发送订阅请求
            doSubscribe(url, listener);
        } catch (Exception e) {
//            notify(url, listener);
            logger.record("Failed to subscribe " + url + ", waiting for retry, cause: " + e.getMessage(), e);
            // 将失败的订阅请求记录到失败列表，定时重试
            addFailedSubscribed(url, listener);
        }
    }

    /**
     * 
     * 解除订阅
     *
     * @author gaojian
     * @created 2013-8-18 上午1:47:15
     *
     * @param url
     * @param listener
     */
    public void unsubscribe(URL url, NotifyListener listener) {
        removeFailedSubscribed(url, listener);
        try {
            // 向服务器端发送取消订阅请求
            doUnsubscribe(url, listener);
        } catch (Exception e) {
            logger.record("Failed to unsubscribe " + url + ", waiting for retry, cause: " + e.getMessage(), e);
            // 将失败的取消订阅请求记录到失败列表，定时重试
            Set<NotifyListener> listeners = failedUnsubscribed.get(url);
            if (listeners == null) {
                failedUnsubscribed.putIfAbsent(url, new HashSet<NotifyListener>());
                listeners = failedUnsubscribed.get(url);
            }
            listeners.add(listener);
        }
    }
    
    /**
     * 
     * 重试
     *
     * @author gaojian
     * @created 2013-8-18 上午1:47:24
     *
     */
    protected void retry() {
        if (null != failedRegisteredURL) {
        	doRegister(failedRegisteredURL);
        	failedRegisteredURL = null;
        }
        if (null != failedUnregisteredURL) {
            doUnregister(failedUnregisteredURL);
            failedUnregisteredURL = null;
        }
        if (!failedSubscribed.isEmpty()) {
            Map<URL, Set<NotifyListener>> failed = new HashMap<URL, Set<NotifyListener>>(failedSubscribed);
            for (Map.Entry<URL, Set<NotifyListener>> entry : new HashMap<URL, Set<NotifyListener>>(failed).entrySet()) {
                if (entry.getValue() == null || entry.getValue().size() == 0) {
                    failed.remove(entry.getKey());
                }
            }
            if (failed.size() > 0) {
                try {
                    for (Map.Entry<URL, Set<NotifyListener>> entry : failed.entrySet()) {
                        URL url = entry.getKey();
                        Set<NotifyListener> listeners = entry.getValue();
                        for (NotifyListener listener : listeners) {
                            try {
                                doSubscribe(url, listener);
                                listeners.remove(listener);
                            } catch (Throwable t) { //忽略所有，等待下次重试
                                logger.record("Failed to retry subscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                            }
                        }
                    }
                } catch (Throwable t) { //忽略所有，等待下次重试
                    logger.record("Failed to retry subscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                }
            }
        }
        if (!failedUnsubscribed.isEmpty()) {
            Map<URL, Set<NotifyListener>> failed = new HashMap<URL, Set<NotifyListener>>(failedUnsubscribed);
            for (Map.Entry<URL, Set<NotifyListener>> entry : new HashMap<URL, Set<NotifyListener>>(failed).entrySet()) {
                if (entry.getValue() == null || entry.getValue().size() == 0) {
                    failed.remove(entry.getKey());
                }
            }
            if (failed.size() > 0) {
                try {
                    for (Map.Entry<URL, Set<NotifyListener>> entry : failed.entrySet()) {
                        URL url = entry.getKey();
                        Set<NotifyListener> listeners = entry.getValue();
                        for (NotifyListener listener : listeners) {
                            try {
                                doUnsubscribe(url, listener);
                                listeners.remove(listener);
                            } catch (Throwable t) { // 忽略所有异常，等待下次重试
                                logger.record("Failed to retry unsubscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                            }
                        }
                    }
                } catch (Throwable t) { // 忽略所有异常，等待下次重试
                    logger.record("Failed to retry unsubscribe " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                }
            }
        }
        if (!failedNotified.isEmpty()) {
            Map<URL, Map<NotifyListener, Set<URL>>> failed = new HashMap<URL, Map<NotifyListener, Set<URL>>>(failedNotified);
            for (Map.Entry<URL, Map<NotifyListener, Set<URL>>> entry : new HashMap<URL, Map<NotifyListener, Set<URL>>>(failed).entrySet()) {
                if (entry.getValue() == null || entry.getValue().size() == 0) {
                    failed.remove(entry.getKey());
                }
            }
            if (failed.size() > 0) {
                try {
                    for (Map<NotifyListener, Set<URL>> values : failed.values()) {
                        for (Map.Entry<NotifyListener, Set<URL>> entry : values.entrySet()) {
                            try {
                                NotifyListener listener = entry.getKey();
                                Set<URL> urls = entry.getValue();
                                listener.notify(urls);
                                values.remove(listener);
                            } catch (Throwable t) { // 忽略所有异常，等待下次重试
                            	logger.record("Failed to retry notify " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                            }
                        }
                    }
                } catch (Throwable t) { // 忽略所有异常，等待下次重试
                	logger.record("Failed to retry notify " + failed + ", waiting for again, cause: " + t.getMessage(), t);
                }
            }
        }
    }
    
    /**
     * 
     * 销毁
     *
     * @author gaojian
     * @created 2013-8-18 上午1:47:36
     *
     */
    public void destroy() {
        try {
            retryFuture.cancel(true);
        } catch (Throwable t) {
            logger.error(t.getMessage(), t);
        }
    }
    
    /**
     * 
     * 通知客户端服务发生变化，一个agent对应多个collector
     *
     * @author gaojian
     * @created 2013-8-18 上午11:56:54
     *
     * @param agentUrl	一个agent
     * @param listener
     * @param collectorUrls	多个collector
     */
    public void notify(URL agentUrl, NotifyListener listener, Set<URL> collectorUrls) {
        try {
        	Map<String, Set<URL>> result = new HashMap<String, Set<URL>>();
        	for (URL u : collectorUrls) {
        		String side = u.getSide();
        		Set<URL> categoryList = result.get(side);
        		if (categoryList == null) {
        			categoryList = new HashSet<URL>();
        			result.put(side, categoryList);
        		}
        		categoryList.add(u);
        	}
        	if (result.size() == 0) {
        		return;
        	}
        	Map<String, Set<URL>> categoryNotified = notified.get(agentUrl);
        	if (categoryNotified == null) {
        		notified.putIfAbsent(agentUrl, new ConcurrentHashMap<String, Set<URL>>());
        		categoryNotified = notified.get(agentUrl);
        	}
        	for (Map.Entry<String, Set<URL>> entry : result.entrySet()) {
        		String category = entry.getKey();
        		Set<URL> categoryList = entry.getValue();
        		categoryNotified.put(category, categoryList);
        		listener.notify(categoryList);
        	}
        } catch (Exception t) {
            // 将失败的通知请求记录到失败列表，定时重试
            Map<NotifyListener, Set<URL>> listeners = failedNotified.get(agentUrl);
            if (listeners == null) {
                failedNotified.putIfAbsent(agentUrl, new ConcurrentHashMap<NotifyListener, Set<URL>>());
                listeners = failedNotified.get(agentUrl);
            }
            listeners.put(listener, collectorUrls);
            logger.record("Failed to notify for subscribe " + agentUrl + ", waiting for retry, cause: " + t.getMessage(), t);
        }
    }
    
}
